package baiduschPro;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.ByteArrayOutputStream;
import org.jsoup.Connection;
import org.jsoup.Jsoup;
import org.jsoup.Connection.Response;
import org.jsoup.nodes.Document;

@Tags({"data post http"})
@CapabilityDescription("data post http")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class HylandaPOSTHTTPProcessor extends AbstractProcessor {

	public static final PropertyDescriptor baseurl = new PropertyDescriptor
            .Builder().name("baseurl")
            .displayName("base url")
            .description("baseurl")
            .required(true)
            .defaultValue("http://")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
	
	public static final PropertyDescriptor otherparams = new PropertyDescriptor
            .Builder().name("otherparams")
            .displayName("otherparams")
            .description("send post parasm string")
            .required(false)
            .defaultValue("other")
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .build();
	  
	private List<PropertyDescriptor> descriptors;
	
	 @Override
	    protected void init(final ProcessorInitializationContext context) {
	        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	        descriptors.add(baseurl);
	        descriptors.add(otherparams);
	        this.descriptors = Collections.unmodifiableList(descriptors);

	    }


	    @Override
	    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
	        return descriptors;
	    }
	
	
	public HylandaPOSTHTTPProcessor() {
		// TODO Auto-generated constructor stub
	}

	 @OnScheduled
    public void onScheduled(final ProcessContext context) {

  //  	System.out.println("onScheduled is clled");
    }
	    
	 
	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
		FlowFile flowFile = session.get();
        if ( flowFile == null ) {
        	
        	System.out.println("flowfile is null");
           return;
        }
        //获取配置属性值
        PropertyValue baseurlp = context.getProperty("baseurl");
        String baseurl = baseurlp.getValue();
        
        PropertyValue otherparamsp = context.getProperty("otherparams");
        String otherparams = otherparamsp.getValue();
        if(otherparams == null)
        	otherparams = "";
        
        String content = "";
        InputStream in = session.read(flowFile);
        try {
			byte[] by = readStream(in);
			content = new String(by,"UTF-8");
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
        
        //把内容编码，POST到接口中，data=
        Map data = new HashMap();
        data.put("data", content);
        if(!otherparams.isEmpty())
        {
        	Map otherdata = paeseOtherParams(otherparams);
        	if(otherdata.size() > 0)
        		data.putAll(otherdata);
        }
        String ret = postdataHttp(baseurl,data);
        
        session.remove(flowFile);
	}
	
	
 public static byte[] readStream(InputStream inStream) throws Exception {  
        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
        byte[] buffer = new byte[1024];  
        int len = -1;  
        while ((len = inStream.read(buffer)) != -1) {  
            outSteam.write(buffer, 0, len);  
        }  
        outSteam.close();  
        inStream.close();  
        return outSteam.toByteArray();  
    } 
 
 //格式：test=23&reuy=yt&eury
 	private Map paeseOtherParams(String params)
 	{
 		Map mapret = new HashMap();
 		params = params.trim();
 		String[] szparams = params.split("&");
 		for(int i = 0 ; i < szparams.length; i++)
 		{
 			String one = szparams[i];
 			String[] onesz = one.split("=");
 			if(onesz.length ==2)
 			{
 				String key = onesz[0];
 				String value = onesz[1];
 				mapret.put(key, value);
 			}
 		}//for i
 		
 		return mapret;
 	}
 
	private  String postdataHttp(String baseurl,Map data)
	{//下载一个正文
		String strpage = "";
		int nloopcnt = 0;
		while(nloopcnt <= 3)
		{
			Connection con = Jsoup.connect(baseurl);
			con.timeout(10*1000);
			con.header("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8");
			con.header("Accept-Encoding","gzip, deflate, sdch");
			con.header("Accept-Language","h-CN,zh;q=0.8");
	//		con.header("Host","www.qichacha.com");
	//		con.header("Referer",referurl);
			con.header("User-Agent","Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.84 Safari/537.36");
			con.data(data);
			con.ignoreContentType(true);
			try 
			{
				Document doc = con.post();
				Response re = con.response();
				
				int ncode = re.statusCode();
				if(ncode == 200)
				{
					strpage = re.body();
					break;
				}
				else if((ncode >=100 && ncode <200)||(ncode >= 500 && ncode < 600))
				{//重试
					nloopcnt++;
					//String strMsg = formatLogInfo("","downLoadOneContent down to try","","httpcode="+ncode,"");
					
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					continue;
				}
				
			}catch (IOException e) 
			{
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			catch(Exception e1)
			{
				e1.printStackTrace();
			}
			nloopcnt++;
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}//while
		return strpage;
	}


}
